//package com.ruoyi.flink.wc;
//
//import com.alibaba.fastjson.JSONObject;
//import com.ruoyi.flink.util.write.ClickHouseWrite;
//import com.ruoyi.flink.wc.entity.Student;
//import com.ruoyi.flink.wc.util.RocketMQSource;
//import org.apache.flink.api.common.functions.FlatMapFunction;
//import org.apache.flink.api.java.utils.ParameterTool;
//import org.apache.flink.streaming.api.datastream.DataStreamSource;
//import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.util.Collector;
//
///**
// * @Author syc
// * @Date Created in 10:21 2022/5/9
// * @Description
// */
//public class TestFlinkClickHouseWrite {
//    @SuppressWarnings("serial")
//	public static void main(String[] args) throws Exception {
//
//        // System.setProperty("rocketmq.client.logLevel", "ERROR");
//        // 参数中获取值
//        final ParameterTool parameterTool = ParameterTool.fromArgs(args);
//        // 获取 rocketmq的地址端口  "192.168.1.125:9876,192.168.1.125:9875"
//        String namesrvaddr = parameterTool.get("namesrvaddr", "192.168.1.123:9876");
//        // 获取主题(topic)
//        String topic = parameterTool.get("topic", "flink-test");
//        // 获取分组(group)
//        String group = parameterTool.get("flinkGroup", "flink-group");
//        // 创建流式执行环境
//        StreamExecutionEnvironment env =  StreamExecutionEnvironment.getExecutionEnvironment();
//        // 设置并行度(默认就是以)
//        env.setParallelism(1);
//        // 创建表环境
//        //  StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//        // 获取mq中的参数值
//        DataStreamSource<String> mqSource = env.addSource(new RocketMQSource(namesrvaddr, topic, group));
//
//        SingleOutputStreamOperator<Student> wordAndOne =  mqSource.flatMap(new FlatMapFunction<String, Student>() {
//            @Override
//            public void flatMap(String sourc, Collector<Student> out) {
//                JSONObject json = JSONObject.parseObject(sourc);
//                Integer age = json.getInteger("age");
//                Student stu = new Student();
//                stu.age = age;
//                stu.sex = "男";
//                stu.name = "王长江";
//                System.out.println(stu.toString() + "...............");
//                out.collect(stu);
//            }
//        });
//        //写入Mysql
//        String sql = "INSERT INTO default.student ( name, age,sex) VALUES (?,?,?)";
//
//        wordAndOne.addSink(new ClickHouseWrite(sql));
//
//        env.execute();
//    }
//
//}
